package com.breeze.blockchain.socket.stomp;

import java.lang.reflect.Type;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompFrameHandler;
import org.springframework.messaging.simp.stomp.StompHeaders;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandler;
import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.messaging.WebSocketStompClient;

/**
 * @author breeze
 */
public class WebSocketClient {

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    WebSocketStompClient stompClient = new WebSocketStompClient(new StandardWebSocketClient());
    stompClient.setMessageConverter(new MappingJackson2MessageConverter());
    stompClient.setTaskScheduler(new ThreadPoolTaskScheduler());
    StompSessionHandler sessionHandler = new MyStompSessionHandler();
    CompletableFuture<StompSession> session =
        stompClient.connectAsync("ws://localhost:8080/stomp", sessionHandler);
    StompSession stompSession = session.get();
    new Timer()
        .schedule(
            new TimerTask() {
              @Override
              public void run() {
                stompSession.send(
                    "/app/hello", new HelloMessage("world" + System.currentTimeMillis()));
              }
            },
            0,
            1000);
  }

  @Slf4j
  public static class MyStompSessionHandler extends StompSessionHandlerAdapter {

    @Override
    public void afterConnected(StompSession session, @NonNull StompHeaders connectedHeaders) {
      session.subscribe(
          "/topic/greetings",
          new StompFrameHandler() {
            @Override
            @NonNull
            public Type getPayloadType(@NonNull StompHeaders headers) {
              return Greeting.class;
            }

            @Override
            public void handleFrame(@NonNull StompHeaders headers, Object payload) {
              Greeting greeting = (Greeting) payload;
              System.out.println(greeting.getContent());
            }
          });
      session.send("/app/hello", new HelloMessage("客户端发送请求"));
    }

    @Override
    public void handleException(
        @NonNull StompSession session,
        StompCommand command,
        @NonNull StompHeaders headers,
        byte @NonNull [] payload,
        Throwable exception) {
      System.out.println(exception.getMessage());
    }

    @Override
    public void handleTransportError(@NonNull StompSession session, @NonNull Throwable exception) {
      log.error("连接错误", exception);
    }
  }
}
